package xyz.chaoslolomonica.recipes;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.data.Stat;

import java.util.List;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

public class ZKDistributedLockImpl implements DistributedLock {

	private CuratorFramework client;
	private String rootPath;
	private String currentPath;
	private volatile boolean locked = false;
	private ReentrantLock lock;
	private Condition previousChangeCondition;

	public static DistributedLock create(CuratorFramework client, String rootPath, boolean fair) {
		return fair ? new ZKDistributedLockImpl(client, rootPath).new FairLock(client, rootPath)
				: new ZKDistributedLockImpl(client, rootPath).new UnfairLock(client, rootPath);
	}

	private ZKDistributedLockImpl(CuratorFramework client, String rootPath) {
		this.client = client;
		this.rootPath = rootPath;
		this.lock = new ReentrantLock();
		previousChangeCondition = this.lock.newCondition();
		try {
			Stat stat = this.client.checkExists().forPath(rootPath);
			if (stat == null) {
				this.client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(rootPath);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void lock() {
		throw new UnsupportedOperationException("please use sub class method");
	}

	@Override
	public void unlock() {
		throw new UnsupportedOperationException("please use sub class method");
	}

	private class MyWatcher implements CuratorWatcher {
		private CountDownLatch latch;
		public MyWatcher(CountDownLatch latch) {
			this.latch = latch;
		}

		@Override
		public void process(WatchedEvent event) throws Exception {
			if (event.getType() == Watcher.Event.EventType.NodeDeleted) {
				this.latch.countDown();
			}
		}
	}

	private class UnfairLock extends ZKDistributedLockImpl {

		private UnfairLock(CuratorFramework client, String rootPath) {
			super(client, rootPath);
		}

		@Override
		public void lock() {
			try {
				String realPath = client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(rootPath + "/l");
				List<String> childrenList = client.getChildren().forPath(rootPath);
				TreeSet<String> treeSet = new TreeSet(
						childrenList.stream().map(l -> rootPath + "/" + l).collect(Collectors.toList()));
				String smallOne = treeSet.first();
				if (realPath.equalsIgnoreCase(smallOne)) {
					currentPath = smallOne;
					return;
				}
				locked = true;
				String lowerOne = treeSet.lower(realPath);
				// register once
				CountDownLatch latch = new CountDownLatch(1);
				Stat stat = client.checkExists().usingWatcher(new MyWatcher(latch)).forPath(lowerOne);
				if (stat != null) {
					latch.await();
				}
				currentPath = realPath;
			} catch (Exception e) {
				e.printStackTrace();
			}
		}

		@Override
		public void unlock() {
			try {
				if (currentPath != null) {
					client.delete().forPath(currentPath);
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	private class FairLock extends ZKDistributedLockImpl {

		private FairLock(CuratorFramework client, String rootPath) {
			super(client, rootPath);
		}

		@Override
		public void lock() {
			try {
				String path = null;
				do {
					try {
						path = client.create().withMode(CreateMode.PERSISTENT).forPath(rootPath + "/l");
					} catch (KeeperException.NodeExistsException e) {
						CountDownLatch latch = new CountDownLatch(1);
						Stat stat = client.checkExists().usingWatcher(new MyWatcher(latch)).forPath(rootPath + "/l");
						if (stat != null) {
							latch.await();
						}
					}
				} while (path == null);
				currentPath = path;
			} catch (Exception e) {
				e.printStackTrace();
			}
		}

		@Override
		public void unlock() {
			try {
				if (currentPath != null) {
					client.delete().forPath(currentPath);
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
}
